Forge - Python 自托管 LLM 工具调用与多步 Agent 框架

from forge import Agent, ClaudeModel, OllamaModel

# 用 Claude
agent_claude = Agent(model=ClaudeModel.SONNET, api_key="sk-...")

# 切换到 Ollama，代码不用改
agent_local = Agent(model=OllamaModel.LLAMA3_70B, base_url="http://localhost:11434")

# 同一个 tool 定义，两个后端都能用
@agent_claude.tool()
@agent_local.tool()
def search_docs(query: str) -> str:
    """从知识库检索相关文档"""
    return vector_db.search(query)

from forge import Agent, RagTool

agent = Agent(model="claude-3-5-sonnet")

# 配置 RAG 工具
rag = RagTool(
    docs_dir="./knowledge_base",
    embedding_model="sentence-transformers/all-MiniLM-L6-v2",
    top_k=5
)

@agent.tool()
def answer_from_docs(question: str) -> str:
    """从私有知识库回答问题"""
    results = rag.search(question)
    context = "\n\n".join([r.text for r in results])
    return agent.chat(f"基于以下上下文回答问题：\n{context}\n\n问题：{question}")

# 使用
answer = answer_from_docs("公司的报销流程是什么？")

from forge import Agent, Chain

agent = Agent(model="claude-3-5-sonnet")

# 定义多步链
chain = Chain([
    # 步骤 1：搜索文档
    agent.create_step(
        name="search_docs",
        tool=rag.search,
        args={"query": "{$user_query}"}
    ),
    # 步骤 2：生成摘要
    agent.create_step(
        name="write_summary",
        prompt="基于以下搜索结果生成摘要：\n{search_docs.result}",
        depends_on=["search_docs"]
    ),
    # 步骤 3：发送邮件
    agent.create_step(
        name="send_email",
        tool=email_sender.send,
        args={
            "to": "team@company.com",
            "subject": "查询结果",
            "body": "{write_summary.result}"
        },
        depends_on=["write_summary"]
    )
])

# 执行
result = chain.run(user_query="Q3 技术架构文档")
print(result["send_email"].output)  # 最终输出

from forge import CodeExecutor, SandboxConfig

# 配置沙箱
sandbox = SandboxConfig(
    timeout=30,          # 执行超时（秒）
    memory_limit="512M", # 内存上限
    network="disabled",  # 禁用网络（可选）
    allowed_modules=["numpy", "pandas", "matplotlib"]  # 白名单模块
)

executor = CodeExecutor(sandbox=sandbox)

@agent.tool()
def run_analysis(code: str) -> str:
    """执行数据分析代码"""
    result = executor.run(code)
    return result.output

pip install forge-ai

from forge import Forge, ClaudeModel

# 初始化
forge = Forge(api_key="your-claude-key")

# 定义工具
@forge.tool()
def search_kb(query: str) -> str:
    """从私有知识库检索"""
    return forge.rag.search(query)

# 定义 Agent
@forge.agent(model=ClaudeModel.SONNET)
def research_task(task: str):
    results = search_kb(task)
    return forge.summarize(results)

# 运行
result = research_task("查找 Q3 技术架构文档")
print(result)

from forge import Forge, OllamaModel

# 指向本地 Ollama 实例
forge = Forge(
    model=OllamaModel.LLAMA3_70B,
    base_url="http://localhost:11434"
)

# 后续代码和 Claude 版本完全一致
@forge.tool()
def search_kb(query: str) -> str:
    # ...

import os
from forge import Forge, ClaudeModel, OllamaModel

MODEL_BACKEND = os.getenv("MODEL_BACKEND", "claude")

if MODEL_BACKEND == "claude":
    forge = Forge(model=ClaudeModel.SONNET, api_key=os.getenv("ANTHROPIC_API_KEY"))
else:
    forge = Forge(model=OllamaModel.LLAMA3_70B, base_url="http://localhost:11434")

from forge.tools import RagTool

rag = RagTool(
    docs_dir="./docs",                          # 文档目录
    embedding_model="sentence-transformers/all-MiniLM-L6-v2",  # 嵌入模型
    chunk_size=512,                             # 分块大小
    chunk_overlap=50,                           # 重叠 token 数
    top_k=5,                                    # 返回 top 5 结果
    rerank=True                                 # 启用重排序
)

# 索引文档（首次运行或文档更新时）
rag.index()

# 搜索
results = rag.search("如何申请年假？")
for r in results:
    print(r.text, r.score)  # 文本和相似度分数

from forge.tools import CodeExecutor

executor = CodeExecutor(
    timeout=60,
    memory_limit="1G",
    network="enabled",       # 允许网络（慎用）
    allowed_modules=["numpy", "pandas", "matplotlib", "seaborn"]
)

@forge.tool()
def analyze_csv(file_path: str, question: str) -> str:
    """分析 CSV 文件"""
    code = f"""
import pandas as pd
df = pd.read_csv('{file_path}')
# 用户问题：{question}
# 生成分析代码...
"""
    result = executor.run(code)
    return result.output

from forge.tools import WebBrowser

browser = WebBrowser(
    headless=True,       # 无头模式
    timeout=30,
    user_agent="Forge/1.0"
)

@forge.tool()
def fetch_webpage(url: str) -> str:
    """抓取网页内容"""
    page = browser.get(url)
    return page.markdown  # 返回 Markdown 格式

from forge import Chain, Condition

chain = Chain([
    agent.create_step("check_intent", prompt="判断用户意图：技术/商务/其他"),
    agent.create_step(
        "tech_response",
        prompt="技术回答：{check_intent.result}",
        condition=Condition.equals("check_intent.result", "技术")
    ),
    agent.create_step(
        "biz_response",
        prompt="商务回答：{check_intent.result}",
        condition=Condition.equals("check_intent.result", "商务")
    )
])

chain = Chain([
    agent.create_step("search", tool=google_search, args={"query": "{$query}"}),
    agent.create_step(
        "should_continue",
        prompt="搜索结果是否满足需求？如果否，返回新的搜索词。如果是，返回 DONE。",
        max_iterations=5  # 最多循环 5 次
    )
])

from forge import RetryPolicy

chain = Chain(
    steps=[...],
    retry_policy=RetryPolicy(
        max_retries=3,
        backoff="exponential",  # 指数退避
        on_failure="pause"      # 失败暂停，等待人工介入
    )
)

from forge import Forge, UserSandbox

forge = Forge()

# 每个用户独立的沙箱
def handle_user_request(user_id: str, query: str):
    sandbox = UserSandbox(
        user_id=user_id,
        work_dir=f"./sandboxes/{user_id}",
        quota={"cpu": 2, "memory": "2G", "disk": "10G"}
    )
    agent = forge.create_agent(sandbox=sandbox)
    return agent.run(query)

from forge import KeyVault

vault = KeyVault(
    backend="env",  # 从环境变量读取
)

forge = Forge(api_key=vault.get("ANTHROPIC_API_KEY"))

from flask import Flask, request, jsonify
from forge import Forge, OllamaModel, RagTool

app = Flask(__name__)
forge = Forge(model=OllamaModel.LLAMA3_70B)
rag = RagTool(docs_dir="./internal_docs")
rag.index()

@forge.agent()
def answer_question(question: str) -> str:
    results = rag.search(question)
    context = "\n\n".join([r.text for r in results])
    return forge.chat(f"上下文：\n{context}\n\n问题：{question}")

@app.route("/ask", methods=["POST"])
def ask():
    question = request.json["question"]
    answer = answer_question(question)
    return jsonify({"answer": answer})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)