企业大语言模型应用：智能体框架之间的技术选型（下）

# 定义工具列表和模型tools = [generate_and_run_sql_query, data_analyzer]model = ChatOpenAI(model="gpt-4", temperature=0).bind_tools(tools)
# 创建智能体图def create_agent_graph():    workflow = StateGraph(MessagesState)
    # 创建工具节点    tool_node = ToolNode(tools)    workflow.add_node("agent", call_model)    workflow.add_node("tools", tool_node)
    # 添加边    workflow.add_edge(START, "agent")    workflow.add_conditional_edges(        "agent",        should_continue,    )    workflow.add_edge("tools", "agent")
    # 添加内存保存器    checkpointer = MemorySaver()    app = workflow.compile(checkpointer=checkpointer)    return app

# 定义是否继续的条件函数def should_continue(state: MessagesState):    messages = state["messages"]    last_message = messages[-1]    if last_message.tool_calls:        return "tools"    return END
# 定义模型调用函数def call_model(state: MessagesState):    messages = state["messages"]    response = model.invoke(messages)    return {"messages": [response]}

# 定义智能体工作流类class AgentFlow(Workflow):    def __init__(self, llm, timeout=300):        super().__init__(timeout=timeout)        self.llm = llm        self.memory = ChatMemoryBuffer(token_limit=1000).from_defaults(llm=llm)        self.tools = [FunctionTool(            skill_map.get_function_callable_by_name(func),            metadata=ToolMetadata(name=func, description=skill_map.get_function_description_by_name(func))        ) for func in skill_map.get_function_list()]
    # 准备智能体上下文的步骤    @step    async def prepare_agent(self, ev: StartEvent) -> RouterInputEvent:        user_input = ev.input        user_msg = ChatMessage(role="user", content=user_input)        self.memory.put(user_msg)
        chat_history = self.memory.get()        return RouterInputEvent(input=chat_history)

# 定义路由步骤@stepasync def router(self, ev: RouterInputEvent) -> ToolCallEvent | StopEvent:    messages = ev.input
    if not any(isinstance(message, dict) and message.get("role") == "system" for message in messages):        system_prompt = ChatMessage(role="system", content=SYSTEM_PROMPT)        messages.insert(0, system_prompt)
    with using_prompt_template(template=SYSTEM_PROMPT, version="v0.1"):        response = await self.llm.achat_with_tools(            model="gpt-4",            messages=messages,            tools=self.tools,        )
    self.memory.put(response.message)
    tool_calls = self.llm.get_tool_calls_from_response(response, error_on_no_tool_call=False)    if tool_calls:        return ToolCallEvent(tool_calls=tool_calls)    else:        return StopEvent(result=response.message.content)

# 定义工具调用处理步骤@stepasync def tool_call_handler(self, ev: ToolCallEvent) -> RouterInputEvent:    tool_calls = ev.tool_calls
    for tool_call in tool_calls:        function_name = tool_call.tool_name        arguments = tool_call.tool_kwargs        if "input" in arguments:            arguments["prompt"] = arguments.pop("input")
        try:            function_callable = skill_map.get_function_callable_by_name(function_name)        except KeyError:            function_result = "Error: Unknown function call"        else:            function_result = function_callable(arguments)        message = ChatMessage(            role="tool",            content=function_result,            additional_kwargs={"tool_call_id": tool_call.tool_id},        )
        self.memory.put(message)
    return RouterInputEvent(input=self.memory.get())