Akemi

Langchain模块Agent组件常规应用

2025/12/25

agent迭代器AgentExecutor

Agent 迭代器（ AgentExecutor）是 LangChain 中实际执行 Agent 决策循环的引擎

即思考 → 行动 → 观察 → 再思考 → … → 结束

特性 Agent自带迭代器 自定义迭代器
开发难度 ✅ 零配置 🟡 需要编程
标准化 ✅ 符合ReAct框架 🟡 任意设计
错误处理 ✅ 内置完善 🟡 需自行实现
工具集成 ✅ 自动适配 🟡 手动集成
循环控制 🟡 固定模式 ✅ 完全定制
状态管理 ✅ 自动维护 🟡 手动管理
性能优化 🟡 一般优化 ✅ 可深度优化
特殊需求 ❌ 难以满足 ✅ 灵活应对 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
from langchain.agents import initialize_agent, AgentType
from langchain.tools import Tool
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv
load_dotenv()

# 1. 定义两个简单的工具
def say_hello(name):
    """向某人问好"""
    return f"你好，{name}！"

def count_words(text):
    """计算文本中的单词数"""
    return len(text.split())

# 2. 创建工具列表
tools = [
    Tool(name="打招呼", func=say_hello, description="向某人问好"),
    Tool(name="数单词", func=count_words, description="计算文本中的单词数")
]

# 3. 初始化 Agent（使用最简单的 ZERO_SHOT）
llm = ChatDeepSeek(model="deepseek-chat")
agent = initialize_agent(
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    tools=tools,
    llm=llm,
    verbose=True
)

# 4. 使用迭代器逐步执行
question = "先向小明打招呼，然后计算 'hello world' 有几个单词"

print("=== 开始迭代执行 ===")

for step in agent.iter({"input": question}):
    print(f"\n当前步骤: {step}")
    
    # 检查是否是中间步骤（Agent的思考过程）
    if isinstance(step, dict) and "intermediate_step" in step:
        action, result = step["intermediate_step"][0]
        print(f"工具调用: {action.tool}")
        print(f"工具输入: {action.tool_input}")
        print(f"工具结果: {result}")
        
    # 询问是否继续（演示人工干预）
    user_input = input("\n继续下一步吗？(y/n): ")
    if user_input.lower() != 'y':
        print("用户终止执行")
        break

print("=== 执行结束 ===")

# > Entering new None chain...
# Thought: 我需要先执行打招呼这个动作，向小明问好。  
# Action: 打招呼  
# Action Input: 小明  

# Observation: 你好，小明  
# ！
# Thought:
# 当前步骤: {'intermediate_step': [(AgentAction(tool='打招呼', tool_input='小明  \n', log='Thought: 我需要先执行打招呼这个动作，向小明问好。  \nAction: 打招呼  \nAction Input: 小明  \n'), '你好，小明  \n！')]}
# 工具调用: 打招呼
# 工具输入: 小明  

# 工具结果: 你好，小明  
# ！

# 继续下一步吗？(y/n): y
# Thought: 打招呼已完成，接下来需要计算 'hello world' 的单词数。  
# Action: 数单词  
# Action Input: hello world  

# Observation: 2
# Thought:
# 当前步骤: {'intermediate_step': [(AgentAction(tool='数单词', tool_input='hello world  \n', log="Thought: 打招呼已完成，接下来需要计算 'hello world' 的单词数。  \nAction: 数单词  \nAction Input: hello world  \n"), 2)]}
# 工具调用: 数单词
# 工具输入: hello world  

# 工具结果: 2

# 继续下一步吗？(y/n): y
# Thought: 我已经完成了两个步骤：先向小明打了招呼，然后数了“hello world”的单词数，结果是2个单词。现在可以给出最终答案。  
# Final Answer: 已向小明打招呼，并且“hello world”有2个单词。

# > Finished chain.

# 当前步骤: {'output': '已向小明打招呼，并且“hello world”有2个单词。', 'messages': [AIMessage(content='Thought: 我已经完成了两个步骤：先向小明打了招呼，然后数了“hello world”的单词数，结果是2个单词。现在可以给出最终答案。  \nFinal Answer: 已向小明打招呼，并且“hello world”有2个单词。', additional_kwargs={}, response_metadata={})]}

# 继续下一步吗？(y/n): y
# === 执行结束 ===

使用向量数据库Combine agents and vector stores

  • 传统 RAG：直接检索 → 生成答案（固定流程）
  • Agent + Vector Store：Agent 决策 → 可能检索 → 可能计算 → 综合答案（动态流程）

动态检索增强：Agent 可以根据需要实时检索向量数据库中的信息，而不是一次性加载所有上下文

智能工具调用：Agent 自主决定何时检索、检索什么、如何结合检索结果

多步骤推理：可以结合多个检索和其他工具（计算、搜索等）完成复杂任务

减少幻觉：基于检索到的真实文档回答，提高准确性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
from langchain.chains import RetrievalQA
from langchain.document_loaders import WebBaseLoader,TextLoader
from langchain.agents import initialize_agent,Tool,AgentType
from langchain_deepseek import ChatDeepSeek
from langchain import LLMMathChain
from langchain.text_splitter import CharacterTextSplitter
from langchain_huggingface import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from dotenv import load_dotenv
load_dotenv()
llm = ChatDeepSeek(model="deepseek-chat")

# 加载文档
loader = TextLoader("./story.txt")
documents = loader.load()
# 文档分割
text_splitter = CharacterTextSplitter(chunk_size=200,chunk_overlap=10)
texts = text_splitter.split_documents(documents=documents)

# 初始化嵌入式模型
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-mpnet-base-v2",
    model_kwargs={'device': 'cpu'}
)
# 创建向量数据库
docsseaech = Chroma.from_documents(texts,embeddings,collection_name="story")

# 创建链
story = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=docsseaech.as_retriever()
)

loader = WebBaseLoader("https://docs.astral.sh/ruff/")

docs = loader.load()
ruff_texts = text_splitter.split_documents(docs)
ruff_db = Chroma.from_documents(ruff_texts,embeddings,collection_name="ruff")
ruff = RetrievalQA.from_chain_type(
    llm=llm,chain_type="stuff",retriever=ruff_db.as_retriever()
)

tools = [
    Tool(name="story",func=story.run,description="对story的查找"),
    Tool(name="RuffQASystem",func=ruff.run,description="回答对于ruff相关问题查找")
]

agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

agent.invoke({"input":"ruff能够帮助艾琳拯救人类吗？"})

# > Entering new AgentExecutor chain...
# Thought: 这个问题涉及Ruff的能力和故事背景，我需要先了解Ruff是什么，以及它在故事中扮演的角色。  
# Action: story  
# Action Input: 查找关于Ruff和艾琳的故事信息
# Observation: 根据您提供的上下文，故事的主角是**艾琳**，但其中并未出现名为 **“Ruff”** 的角色或相关情节。

# 故事主要围绕**艾琳**和她的机器人伙伴 **K-7** 展开，讲述了他们寻找三段星图碎片、对抗神秘组织“黑曜石”，并最终发现关于人类文明真相的冒险。

# 如果您提到的“Ruff”是其他故事或作品中的角色，可能需要更具体的来源信息。在当前提供的科幻故事背景中，没有关于 Ruff 的信息。
# Thought:Thought: 用户问的是“Ruff能否帮助艾琳拯救人类”，但刚才用 story 工具没有找到 Ruff 的相关信息。可能 Ruff 是另一个作品或设定中的角色，或者用户指的是其他东西。既然有一个叫 “RuffQASystem” 的工具，看起来是专门回答关于 ruff 的问题的，我应该用它来查 Ruff 是什么，以及它有什么能力。  
# Action: RuffQASystem  
# Action Input: 查询 Ruff 是什么，以及它有什么能力或功能  

# Observation: 根据提供的上下文，**Ruff** 是一个用 Rust 编写的极快速的 **Python 代码质量工具**，它主要集成了**代码检查（Linting）** 和**代码格式化（Formatting）** 两大核心功能。

# 以下是它的主要能力和功能：

# ### 1. **核心功能**
#    - **代码检查（Linting）**：Ruff 可以快速识别 Python 代码中的错误、风格问题、潜在的 bug 以及不规范的写法。
#    - **代码格式化（Formatting）**：Ruff 可以自动格式化 Python 代码，使其符合统一的风格规范（如 PEP 8）。

# ### 2. **主要特点**
#    - **极快的速度**：由于是用 Rust 编写的，Ruff 的执行速度非常快，远超许多同类工具。用户 Tim Abbott（Zulip 的首席开发者）评价道：“这速度快得离谱……ruff 太棒了。”
#    - **一体化工具**：它集成了代码检查和格式化，无需再单独配置多个工具（如 Flake8、isort、Black 等）。
#    - **易于集成**：Ruff 支持与各种**编辑器**（如 VS Code、PyCharm 等）和开发工作流无缝集成。

# ### 3. **适用场景**
#    - 适用于任何 Python 项目，帮助开发者保持代码质量、一致性和可读性。
#    - 可以用于持续集成（CI）流程，自动检查代码问题。

# ### 总结
# Ruff 是一个**高性能的 Python 代码质量工具**，通过快速的代码检查和格式化功能，帮助开发者提升代码质量和工作效率。如果你正在寻找一个快速、一体化的 Python 代码维护工具，Ruff 是一个很好的选择。
# Thought:Thought: 现在我明白了，Ruff 是一个 Python 代码质量工具（linter/formatter），而艾琳是科幻故事中的人类主角。Ruff 作为现实中的开发工具，无法直接介入虚构故事的情节或帮助艾琳拯救人类。用户的问题可能混淆了两个不同的概念。  
# Final Answer: 不能。Ruff 是一个用 Rust 编写的 Python 代码质量工具（用于代码检查和格式化），而艾琳是科幻故事中的虚构角色。Ruff 作为现实世界的编程工具，不具备在故事中干预情节或拯救人类的能力。

# > Finished chain.

Async API

使用asyncio库为agents提供异步支持

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
from langchain.agents import initialize_agent,load_tools,AgentType
from langchain_deepseek import ChatDeepSeek
from langchain_community.utilities import GoogleSerperAPIWrapper
from langchain.callbacks.stdout import StdOutCallbackHandler
from langchain.callbacks.tracers import LangChainTracer
from aiohttp import ClientSession
from dotenv import load_dotenv
load_dotenv()
import time
import asyncio

llm = ChatDeepSeek(model="deepseek-chat")
tools = load_tools(["google-serper","llm-math"],llm=llm)
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    handle_parsing_errors=True
)

s = time.perf_counter()

questions = [
    "地球到月球的距离是多少千米？这个数据（千米）开根之后的数值是多少",
    "艾菲尔铁塔的高度是多少米？这个数字减去20再除以10的结果是多少？",
    "太阳系中最大的行星是哪一个？它的直径是多少千米？这个数字乘以2再加上1000的结果是多少？"
]
for q in questions:
    agent.run(q)
elapsed = time.perf_counter() - s
print(f"串行执行执行时间: {elapsed:0.2f} 秒.")

s = time.perf_counter()
async def run_agent_async(question: str):
    loop = asyncio.get_event_loop()
    # 在线程池中执行同步的agent.run
    return await loop.run_in_executor(None, agent.run, question)
async def main():
    s = time.perf_counter()
    tasks = [run_agent_async(q) for q in questions]
    results = await asyncio.gather(*tasks)
    
    for q, r in zip(questions, results):
        print(f"Q: {q[:30]}...")
        print(f"A: {r}")
        print("-" * 50)
    
    elapsed = time.perf_counter() - s
    print(f"异步执行时间: {elapsed:0.2f} 秒.")
# 运行
if __name__ == "__main__":
    asyncio.run(main())

# 串行执行执行时间: 107.57 秒.
# 异步执行时间: 63.49 秒.

Custom Multi-Action Agent自定义多动作代理

是 LangChain 中允许一次执行多个工具调用的智能代理。与传统代理（一次只能执行一个动作）不同，它可以在单个推理步骤中规划并执行多个并行操作。

  1. 减少LLM调用次数：一次推理规划多个动作
  2. 并行执行：可同时调用多个API/工具
  3. 复杂任务处理：适合需要多步骤协调的任务
  4. 成本优化：减少API调用次数
方面 传统 initialize_agent (单动作) 自定义 BaseMultiActionAgent (多动作)
决策频率 每执行一个工具，都需要重新决策 一次决策，多个动作
LLM调用次数 每个工具调用一次LLM 减少LLM调用次数
规划时机 实时规划，每个步骤后重新思考 预先批量规划
适用场景 需要逐步推理的复杂任务 可预见的、结构化的多任务
代码复杂度 简单，自动处理 需要自定义规划逻辑 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
from langchain.agents  import Tool,AgentExecutor,BaseMultiActionAgent
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv
load_dotenv()
from langchain_community.utilities import GoogleSerperAPIWrapper

def random_word(query: str) -> str:
    print("\n输出一段文字")
    return "随机文字生成器"

search = GoogleSerperAPIWrapper()
tools = [
    Tool(name="搜索工具", func=search.run, description="用于搜索信息的工具"),
    Tool(name="随机文字生成器", func=random_word, description="生成一段随机的文字")
]

from typing import List, Tuple, Any, Union, Optional
from langchain.schema import AgentAction, AgentFinish

# 自定义agent
class FakeAgent(BaseMultiActionAgent):
    @property
    def input_keys(self) -> List[str]:
        return ["input"]
    def plan(
        self,intermediate_steps: List[Tuple[AgentAction, str]],**kwargs: Any) -> Union[AgentAction, AgentFinish]:
        # 首次调用时返回多个动作
        if len(intermediate_steps) == 0:
            return [
                AgentAction(tool="搜索工具", tool_input=kwargs["input"], log="使用搜索工具"),
                AgentAction(tool="随机文字生成器", tool_input=kwargs["input"], log="使用随机文字生成器")
            ]
        # 有中间步骤后返回完成信号
        else:
            return AgentFinish(return_values={"output": "agent结束"}, log="任务完成")
    def aplan(
        self,intermediate_steps: List[Tuple[AgentAction, str]],**kwargs: Any) -> Union[AgentAction, AgentFinish]:
        if len(intermediate_steps) == 0:
            return [
                AgentAction(tool="搜索工具", tool_input=kwargs["input"], log="使用搜索工具"),
                AgentAction(tool="随机文字生成器", tool_input=kwargs["input"], log="使用随机文字生成器")
            ]
        else:
            return AgentFinish(return_values={"output": "agent结束"}, log="任务完成")
        
agent = FakeAgent()
agent_executor = AgentExecutor.from_agent_and_tools(
    agent=agent,
    tools=tools,
    verbose=True
)

agent_executor.run("2023年世界人口有多少")

# 使用搜索工具世界人口（2025年及历史数据） ; 2023, 8,091,734,930, 0.88% ; 2022, 8,021,407,192, 0.84% ; 2021, 7,954,448,391, 0.86% ; 2020, 7,887,001,292, 1.09% ... 人口, 统计截至号期, 佔世界比, 來源. 世界, 8,026,000,000, 2023年4月15号, 100%, 美國人口普查局-世界人口时钟. 1. 印度, 1,428,000,000, 2023年4月18号, 17.79%, 官方 ... 联合国周四“世界人口日”发布的《2024年世界人口展望》报告指出，到2024 年中期，全球人口达到近82亿，预计在未来60年内还将增加20亿，到2080年代中期达到约 ... 新华社北京12月29日电美国普查局28日公布预估：2024年1月1日世界人口总数将接近80.2亿，比2023年元旦增加超过7516万。 美国普查局在官方网站发表声明说， ... ... 年世界人口展望」（World Population Prospects），預測全球人口將在今年11月15日突破80億人，印度也將在明年取代中國，成為世界上人口最多的國家。 2024年4月29日，世界人口突破81億人。 世界人口自14世紀50年代的黑死病和歐洲大饑荒時期結束後開始迅速增長，當時世界約有3.7億人。 戰爭等因素使增長速度時快時慢。 第二次 ... 2023, 2022, 2021, 2020, 2019, 2018, 2017, 2016, 2015, 2014, 2013, 2012, 2011, 2010, 2009 ... 人口，女性（占总人口的百分比）. 人口总数，女性. 国际移徙者，总计. 據美國普查局數據，2023年全球人口增加7500萬。該機構預計，2025年1月，全球平均每秒鐘出生超過4人、死亡2人。 聯合國網站消息顯示，全球 ... 根据联合国最新报告，全球人口将在今年突破80亿，印度有望在2023年超越中国，成为世界人口最多的国家。不过与此同时，全球人口也正以二战以后、1950年代 ... 新华社北京12月29日电美国普查局28日公布预估：2024年1月1日世界人口总数将接近80.2亿，比2023年元旦增加超过7516万。 美国普查局在官方网站发表声明说， ...使用随机文字生成器
# 输出一段文字
# 随机文字生成器任务完成

原文作者：王盛

原文链接：https://akemi.zj.cn/2025/12/25/Langchain-Agent-app/

发表日期：December 25th 2025, 4:33:08 pm

更新日期：December 25th 2025, 4:34:40 pm

版权声明：本文采用知识共享署名-非商业性使用 4.0 国际许可协议进行许可

CATALOG
  1. 1. agent迭代器AgentExecutor
  2. 2. 使用向量数据库Combine agents and vector stores
  3. 3. Async API
  4. 4. Custom Multi-Action Agent自定义多动作代理
Total : 295
2025
2024
2023
Linux 云原生 云计算 开发 mysql