从零搭建AI智能体:无玄学,仅靠确定性循环

def run_agent(task: str, client: OpenAI, model: str = "gpt-4o-mini") -> str:    messages = [        {            "role": "system",            "content": (                "你是一名智能助手，需要时调用工具；信息完备可给出最终答案时，直接回复正文，不再调用任何工具。"            ),        },        {"role": "user", "content": task},    ]    while True:        response = client.chat.completions.create(            model=model,            messages=messages,            tools=TOOLS,            tool_choice="auto",        )        message = response.choices[0].message        messages.append(message)        # 分支判断：输出答案 或 调用工具        if not message.tool_calls:            return message.content        # 解析并批量执行工具        for tool_call in message.tool_calls:            name = tool_call.function.name            args = json.loads(tool_call.function.arguments)            print(f"  > 正在调用 {name}({args})")            fn = TOOL_FUNCTIONS.get(name)            result = fn(**args) if fn else f"未知工具：{name}"            messages.append({                "role": "tool",                "tool_call_id": tool_call.id,                "content": result,            })        # 单次循环结束，回到while开头新一轮请求

import jsonimport osfrom datetime import datetimefrom openai import OpenAI# 获取当前时间def get_current_date() -> str:    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")# 数学表达式计算def calculate(expression: str) -> str:    try:        result = eval(expression, {"__builtins__": {}}, {})        return str(result)    except Exception as e:        return f"计算异常：{e}"# 模拟天气接口def get_weather(city: str) -> str:    # 可替换为真实天气API请求    return f"{city}天气：72华氏度，局部多云"# 工具函数映射字典TOOL_FUNCTIONS = {    "get_current_date": get_current_date,    "calculate": calculate,    "get_weather": get_weather,}

TOOLS = [    {        "type": "function",        "function": {            "name": "get_current_date",            "description": "获取当前系统日期时间",            "parameters": {"type": "object", "properties": {}, "required": []},        },    },    {        "type": "function",        "function": {            "name": "calculate",            "description": "计算数学表达式结果",            "parameters": {                "type": "object",                "properties": {                    "expression": {                        "type": "string",                        "description": "Python格式数学式，示例：'2 + 2'、'100 * 0.15'",                    }                },                "required": ["expression"],            },        },    },    {        "type": "function",        "function": {            "name": "get_weather",            "description": "查询指定城市实时天气",            "parameters": {                "type": "object",                "properties": {                    "city": {"type": "string", "description": "城市名称"}                },                "required": ["city"],            },        },    },]

if __name__ == "__main__":    client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])    task = "查询今日日期，计算847的15%，再查东京天气"    print(f"任务：{task}\n")    answer = run_agent(task, client)    print(f"\n最终回答：{answer}")

任务：查询今日日期，计算847的15%，再查东京天气  > 正在调用 get_current_date({})  > 正在调用 calculate({'expression': '847 * 0.15'})  > 正在调用 get_weather({'city': 'Tokyo'})最终回答：今日日期2026-04-30 09:14:22，847的15%为127.05；东京气温72华氏度，局部多云。

ollama_client = OpenAI(    base_url="http://localhost:11434/v1",    api_key="ollama" # 客户端必填参数，Ollama本身不校验密钥)answer = run_agent(task, ollama_client, model="qwen2.5")Ollama部署步骤

def ask_cloud_expert(question: str) -> str:    """疑难问题转交云端大模型处理"""    cloud_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])    res = cloud_client.chat.completions.create(        model="gpt-4o",        messages=[{"role": "user", "content": question}]    )    return res.choices[0].message.content

from pathlib import Pathdef web_search(query: str) -> str:    # 模拟搜索接口    return (        f"关键词「{query}」搜索结果：\n"        f"1.维基百科综合词条\n2.五分钟科普短文\n3.官方文档"    )def read_file(path: str) -> str:    return Path(path).read_text()def write_file(path: str, content: str) -> str:    Path(path).write_text(content)    return f"已向{path}写入{len(content)}个字符"# 更新工具映射TOOL_FUNCTIONS = {    "get_current_date": get_current_date,    "calculate": calculate,    "get_weather": get_weather,    "web_search": web_search,    "read_file": read_file,    "write_file": write_file,}

# mcp_server.pyfrom mcp.server.fastmcp import FastMCPmcp = FastMCP("mini-tools")@mcp.tool()def to_uppercase(text: str) -> str:    """文本转大写"""    return text.upper()@mcp.tool()def count_words(text: str) -> int:    """统计单词数量"""    return len(text.split())if __name__ == "__main__":    mcp.run()