LangChain与AutoGPT思想融合：手写Agent循环控制与记忆管理

Python 3.11.5
langchain 0.2.14
langchain-openai 0.1.9
openai 1.40.3
faiss-cpu 1.8.0

while step  3 and last_action == current_action:
        force_switch_tool()  # 防止死循环

from langchain_core.tools import tool
import requests, json, time

@tool
def search_web(query: str) -> str:
    """搜索互联网。query应为2-5个关键词，不要带疑问词。返回前3条结果的标题和URL。"""
    params = {"q": query, "count": 3}
    resp = requests.get("https://api.duckduckgo.com/", params=params, timeout=5)
    data = resp.json()
    results = [f"{item.get('Text', '')[:80]} - {item.get('FirstURL', '')}"
               for item in data.get("RelatedTopics", []) if isinstance(item, dict)]
    return "\n".join(results) if results else "无结果"

@tool
def query_database(table: str, condition: str) -> str:
    """查询SQLite数据库。table可选值: employees, orders。condition是WHERE子句。"""
    import sqlite3
    conn = sqlite3.connect("company.db", timeout=3)
    sql = f"SELECT * FROM {table} WHERE {condition} LIMIT 5"
    try:
        cur = conn.execute(sql)
        rows = cur.fetchall()
        return json.dumps(rows, ensure_ascii=False)
    except Exception as e:
        return f"SQL错误: {e}"
    finally:
        conn.close()

import re, json, time
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferWindowMemory

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, max_tokens=500)
memory = ConversationBufferWindowMemory(k=4, return_messages=True)

SYSTEM_PROMPT = """你是任务规划Agent。请严格输出JSON格式：
{"action": "tool_name", "action_input": {"param": "value"}}
或 {"action": "finish", "action_input": {"answer": "最终答案"}}
可用工具: {tools}"""

def extract_json(text):
    match = re.search(r'\{.*\}', text, re.DOTALL)
    if not match:
        raise ValueError("无JSON块")
    return json.loads(match.group())

def run_agent(task, max_steps=5):
    tools = {"search_web": search_web, "query_database": query_database}
    step = 0
    last_action = None
    repeat_count = 0

    while step  0:
            repeat_count += 1
            if repeat_count >= 2:
                print("[警告] 检测到重复动作，强制切换")
                action = {"action": "search_web", "action_input": {"query": "替代方案"}}
        else:
            repeat_count = 0

        # 执行工具
        if act in tools:
            try:
                result = tools[act].invoke(action["action_input"])
                memory.save_context({"input": prompt}, {"output": f"观察: {result}"})
                print(f"[步骤{step}] {act} -> {str(result)[:50]}...")
            except Exception as e:
                # 错误处理3: 工具执行异常
                err_msg = f"工具{act}执行失败: {str(e)}。请换一个工具。"
                memory.save_context({"input": prompt}, {"output": err_msg})
                print("[错误]", err_msg)
        else:
            print(f"[错误] 未知工具: {act}")

        last_action = act
        step += 1

    if step >= max_steps:
        print("[超时] 达到最大步数，返回当前记忆")
        return memory.load_memory_variables({})

run_agent("找出员工表中工资最高的名字，并在网上搜索该公司的最新动态", max_steps=4)