在本教程中，我们使用 语义的核 与免费的Google Gemini一起，我们在Google Colab上顺利运行。我们首先将语义添加剂注册为工具，例如在Web上搜索，评估数学，输入/输出文件以及注释，然后允许Gemini通过结构化的JSON输出来组织它们。我们看到了代理的计划，通信工具，操作说明和最终答案。核实 完整的代码在这里。
!pip -q install semantic-kernel google-generativeai duckduckgo-search rich
import os, re, json, time, math, textwrap, getpass, pathlib, typing as T
from rich import print
import google.generativeai as genai
from duckduckgo_search import DDGS
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") or getpass.getpass("🔑 Enter GEMINI_API_KEY: ")
genai.configure(api_key=GEMINI_API_KEY)
GEMINI_MODEL = "gemini-1.5-flash"
model = genai.GenerativeModel(GEMINI_MODEL)
import semantic_kernel as sk
try:
from semantic_kernel.functions import kernel_function
except Exception:
from semantic_kernel.utils.function_decorator import kernel_function我们开始安装库和导入基本单元，包括语义内核，双子座和DuckDuckgo搜索。我们准备了API Gemini密钥和模型来创建响应，并准备Kerneel_function从Kerneel记录我们的自定义工具。核实 完整的代码在这里。
class AgentTools:
"""Semantic Kernel-native toolset the agent can call."""
def __init__(self):
self._notes: list[str] = []
@kernel_function(name="web_search", description="Search the web for fresh info; returns JSON list of {title,href,body}.")
def web_search(self, query: str, k: int = 5) -> str:
k = max(1, min(int(k), 10))
hits = list(DDGS().text(query, max_results=k))
return json.dumps(hits[:k], ensure_ascii=False)
@kernel_function(name="calc", description="Evaluate a safe math expression, e.g., '41*73+5' or 'sin(pi/4)**2'.")
def calc(self, expression: str) -> str:
allowed = {"__builtins__": {}}
for n in ("pi","e","tau"): allowed[n] = getattr(math, n)
for fn in ("sin","cos","tan","asin", "sqrt","log","log10","exp","floor","ceil"):
allowed[fn] = getattr(math, fn)
return str(eval(expression, allowed, {}))
@kernel_function(name="now", description="Get the current local time string.")
def now(self) -> str:
return time.strftime("%Y-%m-%d %H:%M:%S")
@kernel_function(name="write_file", description="Write text to a file path; returns saved path.")
def write_file(self, path: str, content: str) -> str:
p = pathlib.Path(path).expanduser().resolve()
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(content, encoding="utf-8")
return str(p)
@kernel_function(name="read_file", description="Read text from a file path; returns first 4000 chars.")
def read_file(self, path: str) -> str:
p = pathlib.Path(path).expanduser().resolve()
return p.read_text(encoding="utf-8")[:4000]
@kernel_function(name="add_note", description="Persist a short note into memory.")
def add_note(self, note: str) -> str:
self._notes.append(note.strip())
return f"Notes stored: {len(self._notes)}"
@kernel_function(name="search_notes", description="Search notes by keyword; returns top matches.")
def search_notes(self, query: str) -> str:
q = query.lower()
hits = [n for n in self._notes if q in n.lower()]
return json.dumps(hits[:10], ensure_ascii=False)
kernel = sk.Kernel()
tools = AgentTools()
kernel.add_plugin(tools, "agent_tools")我们将AgentTools类别指定为我们的语义内核工具组，该工具组为代理商的功能，例如在网络上搜索，安全的数学帐户，检索时间，读取/编写文件以及存储光注释。然后准备语义核并将这些工具记录为附加组件助手，以便代理商在思考时可以打电话给他们。核实 完整的代码在这里。
def list_tools() -> dict[str, dict]:
registry = {}
for name in ("web_search","calc","now","write_file","read_file","add_note","search_notes"):
fn = getattr(tools, name)
desc = getattr(fn, "description", "") or fn.__doc__ or ""
sig = "()" if name in ("now",) else "(**kwargs)"
registry[name] = {"callable": fn, "description": desc.strip(), "signature": sig}
return registry
TOOLS = list_tools()
CATALOG = "\n".join(
[f"- {n}{v['signature']}: {v['description']}" for n,v in TOOLS.items()]
)
SYSTEM = f"""You are a meticulous tool-using AI agent.
You can call TOOLS by returning ONLY a JSON object:
{{"tool":"","args":{{...}}}}
After finishing all steps, respond with:
{{"final_answer":""}}
TOOLS available:
{CATALOG}
Rules:
- Prefer factuality; cite web_search results as A Coding Implementation of an Advanced Tool-Using AI Agent with Semantic Kernel and Gemini(url).
- Keep steps minimal; at most 8 tool calls.
- For file outputs, use write_file and mention the saved path.
- If a tool error occurs, adjust arguments and try again.
"""
def extract_json(s: str) -> dict|None:
for m in re.finditer(r"\{.*\}", s, flags=re.S):
try: return json.loads(m.group(0))
except Exception: continue
return None 我们创建一个List_tools助手，以收集代理记录中的所有可用工具，描述和签名。然后，我们创建了一个列出这些工具并将它们合并到系统系统的目录链，该系统指导Gemini如何以严格的JSON格式连接到工具和最终的抑制。最后，我们定义了extract_json以从表单的形式安全地分析呼叫或最终答案。核实 完整的代码在这里。
def run_agent(task: str, max_steps: int = 8, verbose: bool = True) -> str:
transcript: list[dict] = [{"role":"system","parts":[SYSTEM]},
{"role":"user","parts":[task]}]
observations = ""
for step in range(1, max_steps+1):
content = []
for m in transcript:
role = m["role"]
for part in m["parts"]:
content.append({"text": f"[{role.upper()}]\n{part}\n"})
if observations:
content.append({"text": f"[OBSERVATIONS]\n{observations[-4000:]}\n"})
resp = model.generate_content(content, request_options={"timeout":60})
text = resp.text or ""
if verbose:
print(f"\n[bold cyan]Step {step} - Model[/bold cyan]\n{textwrap.shorten(text, 1000)}")
cmd = extract_json(text)
if not cmd:
transcript.append({"role":"user","parts":[
"Please output strictly one JSON object per your rules."
]})
continue
if "final_answer" in cmd:
return cmd["final_answer"]
if "tool" in cmd:
tname = cmd["tool"]; args = cmd.get("args", {})
if tname not in TOOLS:
observations += f"\nToolError: unknown tool '{tname}'."
continue
try:
out = TOOLS[tname]["callable"](**args)
out_str = out if isinstance(out,str) else json.dumps(out, ensure_ascii=False)
if len(out_str) > 4000: out_str = out_str[:4000] + "...[truncated]"
observations += f"\n[{tname}] {out_str}"
transcript.append({"role":"user","parts":[f"Observation from {tname}:\n{out_str}"]})
except Exception as e:
observations += f"\nToolError {tname}: {e}"
transcript.append({"role":"user","parts":[f"ToolError {tname}: {e}"]})
else:
transcript.append({"role":"user","parts":[
"Your output must be a single JSON with either a tool call or final_answer."
]})
return "Reached step limit. Summarize findings:\n" + observations[-1500:]我们运行代理重复循环，该循环将用户+用户的上下文滋养到双子座，仅强加JSON工具，实现所需的工具，在文本中再次提要注释，返回Final_answer；如果该模型正在从方案中漂流，则我们刺激它，如果它击中了步骤盖，我们将总结结果。核实 完整的代码在这里。
DEMO = (
"Find the top 3 concise facts about Chandrayaan-3 with sources, "
"compute 41*73+5, store a 3-line summary into '/content/notes.txt', "
"add the summary to notes, then show current time and return a clean final answer."
)
if __name__ == "__main__":
print("[bold]🔧 Tools loaded:[/bold]", ", ".join(TOOLS.keys()))
ans = run_agent(DEMO, max_steps=8, verbose=True)
print("\n" + "="*80 + "\n[bold green]FINAL ANSWER[/bold green]\n" + ans + "\n")我们定义了一项试验任务，该任务使代理研究，计算，编写文件，记忆笔记并报告当前时间。然后，我们将代理从一端到末端打开，并打印加载的工具和最终答案，以便我们可以检查完整的工作流程以立即使用工具。
总之，我们注意到内核和双子座是如何在Colab内部进行语义但强大但强大的代理系统的。我们不仅在测试工具调用，而且还看到结果如何再次流向思维环以产生干净的最终答案。现在，我们有一个可重复使用的图表可以扩展更多工具或任务，我们证明，当我们使用正确的框架组合时，构建实用和高级的AI代理可以简单有效。
核实 完整的代码在这里。 毫不犹豫地检查我们的 github页面的课程，符号和笔记本。另外，请随时关注我们 叽叽喳喳 不要忘记加入 100K+ ml子雷迪特 并订阅 我们的新闻通讯。
Asif Razzaq是Marktechpost Media Inc.的首席执行官。作为先驱和视觉工程师，Asif致力于利用人工智能的潜力来促进社会善良。他的最后一项努力是启动人工智能平台Marktechpost，该平台强调了它在机器学习和深度学习新闻的深入覆盖范围内，这在技术上是合理的，可以很容易地被广泛的受众理解。该平台为超过200万个每月观看次数感到自豪，这表明了它在大众中的知名度。
