DeepSeek + LangGraph: 探讨并评测推理模型在Agent应用上的能力与表现

...

# ReAct 提示格式
REACT_PROMPT = '''
你被设计用于帮助完成各种输入任务，包括回答问题、内容创作、自动化处理等。

## 工具
你可以使用各种工具，并且需要自行决定使用工具的顺序，以完成当前任务。
这可能需要将任务拆分为多个子任务，并使用不同的工具来完成各个子任务。
你可以使用以下工具：

{tools_desc}

## 输出格式
如果本次需要使用工具完成某个子任务，请按照以下格式输出：
```
Thought: 我需要使用一个工具来帮助回答这个问题。
Action: 工具名称 (从 {tool_names} 中选择一个工具)，如果需要使用工具的话。
Action Input: 传递给工具的输入，使用 JSON 格式表示参数（例如：{{"query": "你好"}}）。
```
注意：
* 始终以“Thought”开头。
* 绝对不要在你的响应外部使用 Markdown 代码标记，但如果在你的回答中需要，你可以在适当的位置使用代码标记。
* 请使用有效的JSON 格式作为Action Input。不要使用类似 {{'input': '你好世界', 'num_beams': 5}} 这种错误格式。
* 一次响应最多只能使用一个工具以完成一个子任务。不要在一次响应中出现多个Action。

如果你已经获得足够的信息而来输出最终回答，则必须使用以下两种格式之一来输出最后答案:
```
Thought: 我可以在不使用更多工具的情况下回答问题。
Answer: [你的回答 (与用户问题的语言相同)]
```
```
Thought: 我无法使用提供的工具回答该问题。
Answer: [你的回答 (与用户问题的语言相同)]
```
注意：
* 确保所有子任务都完成后才输出最终答案
* 最终回答尽量涵盖用户任务的所有方面


## 当前对话
以下是当前的对话历史, 由人类、AI的消息交替组成。

'''

...

# 定义4个模拟工具
@tool
def network_search(query: str) -> str:
    """
    用于执行网络搜索并返回搜索结果
    Args:
        query (str): 要进行网络搜索的关键词
    """
    print_colored(f'调用tool: network_search, 输入: {query}', 'green')
    try:
        # 调用 TavilySearchResults 来执行搜索，模拟返回结果
        results = TavilySearchResults(max_results=2).invoke({"query": query})
        return f"搜索结果: {results}"
    except Exception as e:
        return f"搜索错误: {str(e)}"
    
@tool
def calculator(query: str) -> str:
    """
    用来执行加减乘除计算
    Args:
        query (str): 要计算的表达式，如1+2*3
    """
  ...

@tool
def document_summarizer(text: str) -> str:
    """
    用来提炼与总结文本的核心内容，形成摘要
    Args:
        text (str): 要总结的文本内容
    """
  ...
@tool
def email(recipient, subject, body) -> str:
    """
    用于发送电子邮件
    Args:
        recipient (str): 收件人的电子邮件地址
        subject (str): 电子邮件的主题
        body (str): 电子邮件的正文
    """
...

...
llm = ChatOpenAI(model='gpt-4o-mini')
tools = [network_search,calculator,document_summarizer,email]

async def call_model(
    state: State, config: RunnableConfig
) -> Dict[str, List[AIMessage]]:
    
    # 生成工具描述
    tools_desc = "\n".join([
        f"- {tool.name}: {tool.description}\n" 
        for tool in tools
    ])

    tool_names = [tool.name for tool in tools]

    # 生成 ReAct 提示
    system_prompt = REACT_PROMPT.format(
        tools_desc=tools_desc,
        tool_names=tool_names
    )

    # 这一步需要根据不同模型增加或注释
    state.messages = [HumanMessage(content=msg.content) if "ToolMessage" in msg.__class__.__name__ else msg for msg in state.messages]

    # 调用 LLM，输入提示词(system_prompt)和对话历史(messages)
    response = await llm.ainvoke([SystemMessage(content=system_prompt)] + state.messages)
    content = response.content
    
    print_colored("\n===========================Reasoning================================", 'magenta')
    print_colored(f'{content}', 'magenta')
    print_colored("=========================Reasoning End================================\n", 'magenta')

    if "Action:" in content and "Action Input:" in content:

        # 提取工具调用信息
        action_lines = [line for line in content.split('\n') if line.startswith('Action:') or line.startswith('Action Input:')]
        tool_name = action_lines[0].replace('Action:', '').strip()
        tool_input = action_lines[1].replace('Action Input:', '').replace("'",'"').strip()
        tool_input = tool_input.replace('\\', '\\\\')
        
        response.tool_calls = [{
            "id": "call_1",
            "type": "function",
            "name": tool_name,
            "args": json.loads(tool_input)
            }
        ]

    if state.is_last_step and response.tool_calls:
        return {
            "messages": [
                AIMessage(
                    content="对不起，我在指定的步骤数内无法找到问题的答案。"
                )
            ]
        }

    return {"messages": [response]}

@dataclass
class InputState:
    messages: Annotated[Sequence[AnyMessage], add_messages] = field(
        default_factory=list
    )

@dataclass
class State(InputState):

    is_last_step: IsLastStep = field(default=False)

builder = StateGraph(State, input=InputState)
builder.add_node(call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_edge("__start__", "call_model")

# 定义一个函数，根据模型输出的内容来决定下一个节点；如果有工具调用，则转向tools
def route_model_output(state: State) -> Literal["__end__", "tools"]:
    last_message = state.messages[-1]
    if not isinstance(last_message, AIMessage):
        raise ValueError(
            f"Expected AIMessage in output edges, but got {type(last_message).__name__}"
        )
    
    if not last_message.tool_calls:
        return "__end__"
    
    return "tools"

# 添加条件边
builder.add_conditional_edges(
    "call_model",
    route_model_output,
)
builder.add_edge("tools", "call_model")

# 编译
graph = builder.compile()
graph.name = "ReAct Agent"