LLM Agent 开发入门

pip3 install langchain

from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from langchain_openai import ChatOpenAI

prompt = PromptTemplate.from_template("写一首关于 {topic} 的诗")
# 使用 ChatOpenAI 替代 OpenAI
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
chain = LLMChain(prompt=prompt, llm=llm)
print(chain.run({"topic": "秋天"}))

from langchain.agents import initialize_agent, load_tools
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
tools = load_tools(["serpapi", "llm-math"], llm=llm)

agent = initialize_agent(
    tools, 
    llm, 
    agent="zero-shot-react-description",
    verbose=True  # 添加详细日志以观察思考过程
)

result = agent.run("中国的首都是哪？它距离南宁有多远？")
print(result)

pip3 install langserve[all]

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langserve import add_routes
from fastapi import FastAPI

prompt = PromptTemplate.from_template("你是谁？回答：{name}")
chain = LLMChain(llm=OpenAI(), prompt=prompt)

app = FastAPI()
add_routes(app, chain, path="/demo")

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langchain_google_genai import ChatGoogleGenerativeAI
import os
import subprocess  # 添加缺失的导入

@tool
def tool_nmap(ip: str) -> str:
    """
    对目标IP执行 nmap 快速扫描
    """
    try:
        # 构建nmap命令：-Pn跳过主机发现，-F进行快速扫描
        cmd = ["nmap", "-Pn", "-F", ip]
        print("[Tool] 执行命令：", " ".join(cmd))
        
        # 执行nmap命令，设置30秒超时
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
        
        # 返回扫描结果或错误信息
        return result.stdout if result.returncode == 0 else result.stderr
    except Exception as e:
        return f"执行 nmap 扫描失败: {e}"

if __name__ == "__main__":
    
    # 创建ReAct代理，集成LLM和nmap工具
    agent = create_react_agent(
        llm,
        tools=[tool_nmap],
    )

    print("请输入 IP 地址开始扫描，或者输入 exit/quit 退出")
    
    # 交互式循环，处理用户输入
    while True:
        user_input = input("请输入目标IP地址: ").strip()
        
        # 检查退出条件
        if user_input.lower() in ["exit", "quit"]:
            break
        
        # 调用代理执行扫描任务
        result = agent.invoke({
            "messages": [
                # 系统角色定义 - 渗透测试工程师
                {"role": "system", "content": "你是一名渗透测试工程师，帮助用户执行信息收集并返回结果。"},
                # 用户输入
                {"role": "user", "content": user_input},
            ]
        })
        
        # 输出扫描结果
        print("扫描结果：")
        print(result['messages'][-1]['content'])

{
	"name": "multiply",
	"args": {
		"a":3,
		"b":7
	},
	type: "tool_call"
}

from langchain_core.tools import tool
import os

@tool
def tool_shell(cmd: str) -> str:
	"""执行 shell 命令"""
	try:
		print("[Tool] 执行命令：",cmd)
		result=os.popen(cmd).read()
		print(f"[Tool] 执行命令 {cmd} 成功，结果：{result}")
		return f"执行命令 {cmd} 成功，结果：{result}"
	except Exception as e:
		return f"执行命令 {cmd} 失败：{e}"

pip install langgraph

from langgraph.prebuilt import create_react_agent

prompt="""
你是一个 shell 命令执行助手，请根据用户的问题执行命令
"""

agent=create_react_agent(
	llm,
	tools=[tool_shell],
	prompt=prompt,
)

response=agent.invoke({"messages":[{"role":"user","content":"当前是什么系统？"}]})

from typing_extensions import TypedDict
from typing import Annotated,List,Tuple

class PlanExecute(TypedDict):
	input: str
	plan: List[str]
	past_steps: Annotated[List[Tuple],operator.add]
	response: str

from typing_extensions import TypedDict
from typing import Annotated,List,Tuple

class PlanExecute(TypedDict):
	input: str              #用户输入
	plan: List[str]         #执行计划步骤列表
	past_steps: Annotated[List[Tuple],operator.add] #已执行步骤记录
	response: str           #最终响应

from pydantic import BaseModel,Field

class Plan(BaseModel):
	"""执行计划"""
	steps: List[str] = Field(
		description="任务列表，应该按顺序执行"
	)
planler_prompt = ChatPromptTemplate.from_messages([
	("system","""
	根据用户输入的任务，生成一个简单的步骤计划。
	这个计划应该涉及单个任务，如果正确执行，将产生正确的答案。
	不要添加任何不必要的步骤。最后一步的结果应该是最终答案。
	确保每一步都要所需的信息 - 不要跳过步骤。
	"""),
	("placeholder","{messages}"),
])

planner = planner_prompt | llm.with_structured_output(Plan)

class Response(BaseModel):
	"""最终响应"""
	response: str

class Act(BaseModel):
	"""下一步行动"""
	action: Union[Response,Plan]=Field(
		description="要执行的动作。如果需要响应用户，使用 Response。"
		"如果需要进一步使用工具获取答案，使用 Plan"
	)

replanner_prompt = ChatPromptTemplate.from_template("""
根据用户输入的任务，分析当前执行情况：

用户任务：{input}
原计划：{plan}
已执行步骤：{past_steps}

请分析：
- 如果某个步骤执行失败，请重新执行该步骤
- 如果需要补充新步骤，请补充计划
- 如果任务已完成，请使用 Response 返回结果
""")

tools = [tool_shell]
prompt = "你是一个 shell 命令执行助手，请根据用户的问题执行命令"
agent_executor = create_react_agent(llm,tools,prompt=prompt)

async def plan_step(state: PlanExecute):
	"""制定初始计划"""
	plan = await planner.ainvoke({"messages": [("user",state[input])]})
	return {"plan":plan.steps}

async def execute_step(state: PlanExecute):
	"""执行当前步骤"""
	task = state["plan"][0] #取第一个未完成的任务
	agent_response = await agent_executor.ainvoke(
		{"messages": [("user",task)]}
	)
	return {
		"past_steps": [(task, agent_response["messages"][-1].content)]
	}
	
async def replan_step(state: PlanExecute):
	"""重新规划"""
	output = await replanner.ainvoke(state)
	if isinstance(output.action, Response):

workflow = StateGraph(PlanExecute)

# 添加节点
workflow.add_node("planner", plan_step)  # 计划节点
workflow.add_node("agent", execute_step) # 执行节点
workflow.add_node("replan", replan_step) # 重规划节点

# 定义边和流程
workflow.add_edge(START, "planner")      # 开始 -> 计划
workflow.add_edge("planner", "agent")    # 开始 -> 计划
workflow.add_edge("agent", "replan")     # 开始 -> 计划

# 条件边：根据重规划结果决定下一步
def should_end(state: PlanExecute):
	if "response" in state and state["response"]:
		return ENO # 任务完成，结束
	else:
		return "agent" # 继续执行

# 编译工作流
app = workflow.compile()

task = """
1. 统计文件系统使用情况
2. 使用 python3 matplotlib 绘制柱状图，并保存为图片（1.png）。

命令的上下文可能比较多，你可以保存到临时文件，然后使用 head、tail 等命令查看结构
请确保有图片输出和文件输出，并告诉我输出了哪些结果文件。
"""

# 执行任务
result = await app.ainvoke({"input": task})

import subprocess

@tool
def tool_shell(cmd: str) -> str:
	"""执行 shell 命令，并返回标准输出或错误信息"""
	try:
		print
		result = subprocess.run(
			cmd, shell=True
			stdout=subprocess.PIPE,
			stderr,subprocess.PIPE,
			text=True
		)
		stdout = result.stdout.strip()
		stderr = result.stderr.strip()

		if result.returncode == 0:
			if stdout:
				return f"执行命令 {cmd} 成功，结果：{stdout}"
			else:
				return f"执行命令 {cmd} 成功，但标准输出为空。"
		else:
			return f"执行命令 {cmd} 失败，错误信息：{stderr}"
	except Exception as e:
		return f"执行命令 {cmd} 异常：{e}"

from typing import Annotated
from langgraph.perbuilt import InjectedState

@tool
def research_agent_too(
	query: str,
	state: Annotated[dict, InjectedState]
) -> str:
	"""调用研究智能体进行网络搜索"""
	response = research_agent.invoke({"messages":[("user",query)]})
	return response["messages"][-1].content

@tool
def python_agent_tool(
	task: str,
	state: Annotated[dict,InjectedState]
) -> str:
	"""调用Python智能体执行代码任务"""
	response = python_agent.invoke({"messages":[("user",task)]})
	return response["messages"][-1].content

# Supervisor 作为 ReAct 智能体
supervisor = create_react_agnet(
	model=llm,
	tools=[research_agent_tool,python_agent_tool],
	prompt="You are a supervisor. Use the available agent tools to complete tasks."
)

# 创建专业团队
data_team = create_supervisor(
	model=llm
	agents=[python_agent, shell_agent, file_agent],
	prompt="管理数据处理相关的智能体团队",
	name="data_team_supervisor"
)

research_team = create_supervisor(
	model=llm
	agents=[search_agent, analysis_agent, summary_agent],
	prompt="管理研究和分析相关的智能体团队",
	name="research_team_supervisor"
)

# 顶级 Supervisor
top_supervisor = create_supervisor(
	model=llm,
	agents=[data_team,research_team],
	prompt="协调各专业团队，处理跨领域复杂任务",
	name="top_supervisor"
)

# 复杂任务：搜索并管理 Android CVE 信息
ans = supervisor.invoke({
	"messages":[
		{
			"role":"user",
			"content":"搜索最新的 5 条 Android CVE，并整理到文件 Android-CVE.md，并为每一个 CVE 创建一个 markdown 文件，最后使用 matplotlib 绘图统计 CVE 对应的 CWE 情况",
		}
	]
})

from langchain_tavily import TavilySearch
import os

os.environ["TAVILY_API_KEY"] = "tvly-......."
web_search = TavilySearch(max_results=3)
print(web_search.invoke({"query": "kali 最新版本"}))

# 研究智能体 - 负责网络搜索
research_agent = create_react_agent(
	model=llm,
	tools[web_search],
	prompt=(
		"You are a research agent.\n\n"
		"INSTRUCTIONS:\n"
		"- Assist ONLY with research-related tasks, DO NOT do any math\n"
		"- After you're aone with your tasks, respond to the supervisor directly\n"
		"- Respond ONLY with the results of your work, do NOT include ANY other text."
	),
	name="research_agent",
)

# Shell 智能体 - 负责系统命令执行
shell_agent = create_react_agent(
	model=llm,
	tools=[tool_shell],
	name="shell_agent",
	prompt=(
		"You are a shell agent.\n\n"
		"INSTRUCTIONS:\n"
		"- Assist ONLY with research-related tasks, DO NOT do any math\n"
		"- After you're aone with your tasks, respond to the supervisor directly\n"
		"- Respond ONLY with the results of your work, do NOT include ANY other text."
	)
)

supervisor = create_supervisor(
	model=llm,
	agents=[shell_agent,research_agent],
	prompt=(
		"You are a supervisor managing three agents:\n"
		"- a shell agent. Assign shell-related tasks to this agent, do not exec python.\n"
		"- a research agent. Assign research-related tasks to this agent, for web search.\n"
		"Assign work to one agent at a time, do not call agents in parallel.\n"
		"Do not use shell agent for Python tasks. \n"
		"Do not do any work yourself."
	),
	add_handoff_back_messages=True,
	output_mode="full_history",
).complie()

from langgraph_supervisor import create_supervisor

vupervisor = create_supervisor(
	model=llm,                       # 用于决策的LLM
	agents=[agent1,agent2,agent3],   # 管理的智能体列表
	prompt="supervisor 的系统提示词", # 调度策略
	add_handoff_back_messages=True,  # 是否添加handoff消息
	output_mode="full_history",      # 输出模式
).compile()

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
	"""Add two numbers"""
	return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
	"""Multiply two numbers"""
	return a * b

if __name__ == "__main__":
	mcp.run(transport="stdio")